﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using Framework.ThreadWorker;
using Framework.Bus;
using System.Messaging;
using System.Threading;

namespace Framework.Event
{
    public class EventProcessor : BaseThreadWorker, IEventProcessor
    {
        IMessageDispatcher _messageDispatcher;
        protected MessageQueue _messageQueue;
        private readonly bool _useInternalTransaction;


        protected virtual void ProcessMessage(Message message)
        {
            if (message != null)
            {
                message.Formatter = _messageQueue.Formatter;
                _messageDispatcher.DispatchMessage(message.Body);
            }
        }

        protected IMessageDispatcher MessageDispatcher
        {
            get
            {
                return _messageDispatcher;
            }
        }

        protected MessageQueue MessageQueue
        {
            get
            {
                return _messageQueue;
            }
        }

        public EventProcessor(IMessageDispatcher messageDispatcher, string msmqPath)
        {
            _messageDispatcher = messageDispatcher;
            try
            {
                var options = new MSMQBusOptions(msmqPath);
                _messageQueue = new MessageQueue(msmqPath,
                    options.SharedModeDenyReceive,
                    options.EnableCache, options.QueueAccessMode);
                _messageQueue.Formatter = new BinaryMessageFormatter(System.Runtime.Serialization.Formatters.FormatterAssemblyStyle.Full,
                    System.Runtime.Serialization.Formatters.FormatterTypeStyle.TypesAlways);
                MaxJoinTimeout = int.MaxValue;
                _messageQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(_messageQueue_ReceiveCompleted);
                this._useInternalTransaction = options.UseInternalTransaction && _messageQueue.Transactional;
              
            }
            catch (System.Exception ex)
            {
                Console.Write(ex.Message);
            }
        }

        public Message[] GetAllMessages()
        {
            return this._messageQueue.GetAllMessages();
        }

        protected virtual void _messageQueue_ReceiveCompleted(object sender, ReceiveCompletedEventArgs asyncResult)
        {
            if (_exit)
            {
                return;
            }
            try
            {
                MessageQueue mq = (MessageQueue)sender;
                // End the asynchronous receive operation.

                Message message = mq.EndReceive(asyncResult.AsyncResult);
                ProcessMessage(message);
            }
            catch (System.Exception ex)
            {
                Console.Write(ex.Message);
            }

        }

        protected virtual void PeekProcess()
        {
            var waitHandle = _messageQueue.BeginPeek();
            waitHandle.AsyncWaitHandle.WaitOne();
        }

        protected virtual void ReveiveProcess()
        {
            var waitHandle = _messageQueue.BeginReceive();
            waitHandle.AsyncWaitHandle.WaitOne();
        }


        protected override void Work()
        {
            try
            {
                ReveiveProcess();
            }
            catch (ThreadInterruptedException)
            {

            }
            catch (System.Exception)
            {

            }
        }

        public override void Stop(bool forcibly = false)
        {
            lock (_mutex)
            {
                _exit = true;
                if (_suspend)
                {
                    Resume();
                }
                if (forcibly)
                {
                    _thread.Abort();
                    _thread = null;
                }
                else
                {
                    _thread.Interrupt();
                    _thread.Join(MaxJoinTimeout);
                    _messageQueue.Close();
                    if (!_thread.ThreadState.Equals(ThreadState.Stopped))
                    {
                        _thread.Abort();
                        _thread = null;
                    }
                }
            }
        }

        protected override int GetWorkInterval()
        {
            return 0;
        }
    }
}